package com.lagou.kafka.demo.producer;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;


import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @param
 * @author oyz
 * @version 1.0
 * @description: 生产者1
 * @date 2022/3/25 15:05
 */
public class Myproducer1 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Map<String,Object> configs = new HashMap<>(3);
        // 连接的broker的地址
        configs.put("bootstrap.servers","192.168.106.130:9092");
        // 指定key的序列化
        configs.put("key.serializer", IntegerSerializer.class);
        // 指定value的序列化
        configs.put("value.serializer", StringSerializer.class);
        //configs.put("acks","all");
        //configs.put("reties","3");

        // 用于设置用户自定义的消息头
        List<Header> headers = new ArrayList<>();
        headers.add(new RecordHeader("biz.name","producer.demo".getBytes()));

        ProducerRecord<Integer,String> record = new ProducerRecord<>("tp_user_01",0,0,"helle lagou0",headers);

        KafkaProducer<Integer,String> producer = new KafkaProducer<Integer, String>(configs);
        // 消息的同步确认的方式
        final Future<RecordMetadata> future = producer.send(record);
        final RecordMetadata recordMetadata = future.get();
        System.out.println("消息的主题："+recordMetadata.topic());
        System.out.println("消息的分区号："+recordMetadata.partition());
        System.out.println("消息的偏移量："+recordMetadata.offset());

        // 消息异步的确认方式
        /*producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                  if(exception == null) {
                      System.out.println("消息的主题："+metadata.topic());
                      System.out.println("消息的分区号："+metadata.partition());
                      System.out.println("消息的偏移量："+metadata.offset());
                  }else{
                      System.out.println("错误的异常信息:"+exception.getMessage());
                  }
            }
        });*/

        // 关闭生产者
        producer.close();


    }
}
